package com.dc.aimc.common.mqtt.config;

import com.dc.aimc.common.mqtt.subscribe.MqttSubscriber;
import lombok.Setter;
import lombok.extern.log4j.Log4j2;
import org.eclipse.paho.client.mqttv3.MqttConnectOptions;
import org.springframework.beans.factory.annotation.Autowired;
import org.springframework.boot.context.properties.ConfigurationProperties;
import org.springframework.context.annotation.Bean;
import org.springframework.context.annotation.Configuration;
import org.springframework.integration.annotation.ServiceActivator;
import org.springframework.integration.channel.DirectChannel;
import org.springframework.integration.core.MessageProducer;
import org.springframework.integration.mqtt.core.DefaultMqttPahoClientFactory;
import org.springframework.integration.mqtt.core.MqttPahoClientFactory;
import org.springframework.integration.mqtt.inbound.MqttPahoMessageDrivenChannelAdapter;
import org.springframework.integration.mqtt.outbound.MqttPahoMessageHandler;
import org.springframework.integration.mqtt.support.DefaultPahoMessageConverter;
import org.springframework.messaging.MessageChannel;
import org.springframework.messaging.MessageHandler;

import java.util.UUID;

@Log4j2
@Setter
@Configuration
@ConfigurationProperties("spring.mqtt")
public class MqttConfig {

    /**
     * 服务器地址 [tcp://182.200.92.86:1883, tcp://182.200.92.86:1883]
     */
    private String[] urls;

    /**
     * 用户名
     */
    private String username;

    /**
     * 密码
     */
    private String password;

    /**
     * 连接超时时间 秒
     */
    private Integer timeout;

    /**
     * 连接心跳间隔 秒
     */
    private Integer keepalive;

    /**
     * 消息质量
     */
    private Integer qos;

    /**
     * 订阅的topic
     */
    private String[] topics;


    @Autowired(required = false)
    private MqttSubscriber mqttSubscriber;

    /**
     * 配置DefaultMqttPahoClientFactory
     * 1. 配置基本的链接信息
     * 2. 配置maxInflight，在mqtt消息量比较大的情况下将值设大
     */
    @Bean
    public MqttPahoClientFactory mqttPahoClientFactory() {
        MqttConnectOptions options = new MqttConnectOptions();
        // 配置mqtt的服务地址 账号 密码 超时 心跳 清除会话 自动重连
        options.setServerURIs(urls);
        options.setUserName(username);
        options.setPassword(password.toCharArray());
        options.setConnectionTimeout(timeout);
        options.setKeepAliveInterval(keepalive);
        options.setCleanSession(false);
        options.setAutomaticReconnect(true);
        // 允许同时发送几条消息
        options.setMaxInflight(200);

        DefaultMqttPahoClientFactory factory = new DefaultMqttPahoClientFactory();
        factory.setConnectionOptions(options);
        return factory;
    }


    /**
     * mqtt消息出站通道，用于发送出站消息
     */
    @Bean
    public MessageChannel messageOutChannel() {
        return new DirectChannel();
    }

    /**
     * mqtt消息出站通道配置
     * 1. 通过MqttPahoMessageHandler初始化出站通道适配器
     * 2. 配置异步发送
     * 3. 配置默认的服务质量
     */
    @Bean
    @ServiceActivator(inputChannel = "messageOutChannel")
    public MqttPahoMessageHandler mqttPahoMessageHandler(MqttPahoClientFactory factory) {
        // 初始化出站通道适配器 Eclipse Paho MQTT客户端库
        String clientId = UUID.randomUUID().toString();
        MqttPahoMessageHandler messageHandler = new MqttPahoMessageHandler(clientId, factory);
        // 异步发送 默认是false(发送时阻塞)
        messageHandler.setAsync(true);
        // 默认的服务质量
        messageHandler.setDefaultQos(qos);
        // 默认topic
        // messageHandler.setDefaultTopic();
        return messageHandler;
    }



    /**
     * mqtt消息入站通道，订阅消息后消息进入的通道。
     * 可创建多个入站通道，对应多个不同的消息生产者。
     *
     * @return
     */
    @Bean
    public MessageChannel messageInChannel() {
        return new DirectChannel();
    }

    /**
     * 配置Inbound入站，消费者基本连接配置
     * 1. 通过DefaultMqttPahoClientFactory 初始化入站通道适配器
     * 2. 配置超时时长，默认30000毫秒
     * 3. 配置Paho消息转换器
     * 4. 配置发送数据的服务质量 0~2
     * 5. 配置订阅通道
     *
     * @return
     */
    @Bean
    public MessageProducer channelInbound(MessageChannel messageInChannel, MqttPahoClientFactory factory) {
        // 订阅topic
        log.info("订阅topic: {}, {}", topics.length, topics);
        String clientId = UUID.randomUUID().toString();
        MqttPahoMessageDrivenChannelAdapter adapter = new MqttPahoMessageDrivenChannelAdapter(clientId, factory, topics);
        adapter.setConverter(new DefaultPahoMessageConverter());
        adapter.setQos(qos);
        adapter.setOutputChannel(messageInChannel);
        return adapter;
    }

    /**
     * mqtt入站消息处理工具，对于指定消息入站通道接收到生产者生产的消息后处理消息的工具。
     *
     * @return
     */
    @Bean
    @ServiceActivator(inputChannel = "messageInChannel")
    public MessageHandler mqttMessageHandler() {
        return mqttSubscriber;
    }

}
